package middleware.p2p.replication;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.PrintStream;
import java.net.MalformedURLException;
import java.net.Socket;
import java.net.URL;
import java.net.UnknownHostException;
import java.util.Iterator;
import java.util.LinkedList;

import javax.ejb.EJB;
import javax.ejb.Singleton;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.xml.parsers.DocumentBuilderFactory;

import middleware.p2p.log.VectorClock;
import middleware.p2p.transaction.wsclient.TransactionCoordinator;
import middleware.p2p.transaction.wsclient.TransactionCoordinatorService;

import org.apache.log4j.Logger;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

/**
 * Session Bean implementation class ReplicaRequestQueue
 */
@Singleton(name="ReplicaQueue")
public class ReplicaRequestQueue implements ReplicaRequestQueueLocal {
	@EJB(name="ejb/ReplicaCache",
			beanInterface=middleware.p2p.replication.ReplicaTransactionCacheLocal.class)
	private ReplicaTransactionCacheLocal replicaCache;
	private LinkedList<ReplicaRequest> incoming;
	private VectorClock vclock;	
	private URL thisWsdl;
	private Logger logger=Logger.getLogger(this.getClass());
    /**
     * Default constructor. 
     */
    public ReplicaRequestQueue() {
        super();
        this.vclock=new VectorClock();
        this.incoming= new LinkedList<ReplicaRequest>();
        try {
	        Context ctx = new InitialContext();
			this.replicaCache = (ReplicaTransactionCacheLocal)ctx.lookup("java:comp/env/ejb/ReplicaCache");
		} catch (NamingException e) {		
		}
		File confFile=new File("replica_manager_config.xml");
		try {
			Document doc = DocumentBuilderFactory.newInstance().newDocumentBuilder().parse(confFile);
			doc.getDocumentElement().normalize();
			NodeList nodeLst = doc.getElementsByTagName("replica_manager");
			if ((nodeLst.getLength()>0)&&(nodeLst.item(0).getNodeType()==Node.ELEMENT_NODE)) {
				Element rmElem=(Element)nodeLst.item(0);
				Element wsdlLoc=(Element)(rmElem.getElementsByTagName("wsdl_url").item(0));
				this.thisWsdl=new URL(((Node)wsdlLoc.getChildNodes().item(0)).getNodeValue());
			}
		} catch (Exception ge) {
			logger.error(ge);
			this.thisWsdl=null;
		} 
    }
    private void deliver(ReplicaRequest event) {
    	Iterator<String> keyIt=event.getVclock().getAllKey().iterator();
    	while (keyIt.hasNext()) {
    		String cKey=keyIt.next();
    		int currValue=this.vclock.getClock(cKey);
    		int remValue=event.getVclock().getClock(cKey);
    		if (remValue>currValue) {
    			this.vclock.setClock(cKey, remValue);
    		} else if((currValue==0)&&(remValue==0)) {
    			this.vclock.setClock(cKey, 0);
    		}
    	}
    	System.out.println(event.getVclock());
    	String resp=null;
    	if (event.isGet()) {
    		resp=this.getReplica(event.getComponentId(), event.getTransId(), event.getCoordWsdl(), event.getVclock());
    	} else {
    		resp=Boolean.toString(this.storeReplica(event.getComponentId(),event.getReplica(), event.getTransId(), event.getCoordWsdl(), event.getVclock()));
    		
    	}
    	resp=(resp!=null?resp:"");
    	String responseAddr[]=event.getResponseAddr().split(":");
    	int port=Integer.parseInt(responseAddr[1]);    	
    	try {
			Socket sock=new Socket(responseAddr[0],port);
			PrintStream ps=new PrintStream(sock.getOutputStream());
			ps.println(resp.replace("\n", "").replace("\r","").replace("\t", ""));
		} catch (NumberFormatException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
    private boolean deliverCompare(ReplicaRequest right) {
    	int comp=vclock.compareTo(right.getVclock());
		boolean gapDetect=vclock.detectGap(right.getVclock());
		boolean proxDetect=vclock.detectProximity(right.getVclock());
		//System.out.println("COMPARE=>"+comp+" || GAP =>"+gapDetect+" || PROX =>"+proxDetect);
		//System.out.println("$$$$$$$$$$$$$$$$$$$$$$$"+vclock.toString()+"\n********************"+right.getVclock().toString());
		if ((comp==-1)&&(!gapDetect)&&(proxDetect)) {
			return true;
		} else if((comp==0)&&(proxDetect)) {
			return true;
		} else {
			return false;
		}
    }
	@Override
	public boolean enqueue(ReplicaRequest event) {
		/*if (intervalTime==null) {
			TimerConfig tConfig=new TimerConfig("Monitor Queue Sorter Timeout",true);
			intervalTime=timerService.createIntervalTimer(60000,1000,tConfig);
			//System.out.println("Timer created");
		}*/
		try{
			if (deliverCompare(event)) {
				//System.out.println("========>immediate delivery \n");
				deliver(event);
				boolean found=true;
				while (found) {
					Iterator<ReplicaRequest> itEv=this.incoming.iterator();
					found=false;
					while (itEv.hasNext()) {
						ReplicaRequest currEvent=itEv.next();
						if (deliverCompare(currEvent)) {
							itEv.remove();
							deliver(currEvent);
							found=true;
							break;
						}
					}					
				}
			} else {
				incoming.add(event);							
			}
			return true;
		}
		catch(IllegalStateException e){
			e.printStackTrace();
			return false;
		}
	}
	@Override
	public boolean resetClock(String cid) {
		System.out.println("Reset Clock "+cid);
		this.vclock.setClock(cid, 0);
		return true;
	}
	private boolean storeReplica(String componentId, String replica,
			String transId, String coordWsdl,VectorClock vc) {
		//System.out.println("STORE");
		URL coordWsdlUrl=null;
		try {
			coordWsdlUrl=new URL(coordWsdl);
		} catch (MalformedURLException mfue) {
			logger.error(mfue);
		}
		if (coordWsdlUrl!=null) {
			//TODO: controllare se fare il join
			//System.out.println(coordWsdlUrl.toString());
			TransactionCoordinatorService service=new TransactionCoordinatorService(coordWsdlUrl);
			TransactionCoordinator coord=service.getTransactionCoordinatorPort();
			coord.join(transId, thisWsdl.toString());
			//coord.abortTransaction(transId);
		}
		logger.debug("storeReplica (inside "+transId+" transaction)");		
		replicaCache.storeReplicaEntry(transId, componentId, replica,ReplicaAction.Store);
		return true;
	}
	private String getReplica(String componentId, String transId, String coordWsdl,VectorClock vc) {
		//this.replicaVCManager.tryDeliver(vc);
		URL coordWsdlUrl=null;
		try {
			coordWsdlUrl=new URL(coordWsdl);
		} catch (MalformedURLException mfue) {
			logger.error(mfue);
		}
		if (coordWsdlUrl!=null) {
			//TODO: controllare se fare il join
			TransactionCoordinatorService service=new TransactionCoordinatorService(coordWsdlUrl);
			TransactionCoordinator coord=service.getTransactionCoordinatorPort();
			coord.join(transId, thisWsdl.toString());
			//coord.abortTransaction(transId);
		}
		logger.debug("getReplica (inside "+transId+" transaction)");
		boolean status=true;
		boolean notFound=false;
		StringBuilder sb=new StringBuilder();
		try {
			BufferedReader inFile = new BufferedReader(new FileReader(new File(".."+File.separator+"logs"+File.separator+"replicaStore"+File.separator+componentId+".rpl")));
			String tLine;
			while ((tLine=inFile.readLine())!=null) {
				sb.append(tLine);
				sb.append("\n");
			}
			inFile.close();
		} catch (FileNotFoundException fnfe) {
			this.logger.error(fnfe);
			notFound=true;
		} catch (IOException ioe) {
			status=false;
			this.logger.error(ioe);
		}
		String content=null;
		if (!notFound&&status) {
			content=sb.toString();
		}
		this.replicaCache.storeReplicaEntry(transId, componentId, String.valueOf(status),ReplicaAction.Get);
		return content;
	}
}
